package com.css.lossHouse.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.baomidou.dynamic.datasource.toolkit.DynamicDataSourceContextHolder;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.css.casualties.entity.Casulaties;
import com.css.casualties.entity.InfluenceYsx;
import com.css.casualties.service.InfluenceYsxService;
import com.css.casualties.service.impl.CasualtiesServiceImpl;
import com.css.config.QueueConfig;
import com.css.config.ThreadPoolConfig;
import com.css.lossHouse.entity.LossHouse;
import com.css.lossHouse.mapper.LossHouseMapper;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * @author Dbj
 * @date 2024年09月27日  15:53
 * 失去住所人数评估
 */
@Service
@Slf4j
public class LossHouseService extends ServiceImpl<LossHouseMapper, LossHouse> {

    @Resource
    InfluenceYsxService influenceYsxService;
    @Autowired
    CasualtiesServiceImpl casualtiesService;
    private static final Map<String, Integer> retryMap = new ConcurrentHashMap<>();
    //失败重试次数
    @Value("${spring.rabbitmq.failedRetry}")
    private Integer failedRetry;
    ThreadPoolExecutor executor = ThreadPoolConfig.exeBizThreadPoolExecutor;

    @RabbitListener(queues={QueueConfig.QUEUE_CASULATIES_TO_LOSSHOUSE}, ackMode = "MANUAL")
    @RabbitHandler
    @DSTransactional
    public void evaluate(Map<String,Object> map, Channel channel, Message message) throws IOException {
        CompletableFuture completableFuture = CompletableFuture.runAsync(()->{
            log.info("开始评估失去住所人数");
            try {
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            } catch (IOException e) {
                e.printStackTrace();
            }
            //地震id
            String earthQuakeId = map.get("earthQuakeId").toString();
            log.error("地震id"+earthQuakeId+"-- 评估失去住所人数"+"|| 开始。>> 参数:"+JSON.toJSONString(map));
            //影响场版本
            String influenceVersion = JSONObject.parseObject(map.get("influenceVersion").toString(),String.class);
            try{
                //数据源
                String dataSourceKey = map.get("dataSource").toString();
                DynamicDataSourceContextHolder.push(dataSourceKey);
                List<LossHouse> resultList = new ArrayList<>();
                //获取人员伤亡的结果
                LambdaQueryWrapper<Casulaties> casulatieslqw = new LambdaQueryWrapper<>();
                casulatieslqw.eq(Casulaties::getEarthQuakeId,earthQuakeId).eq(Casulaties::getVersion,influenceVersion);
                List<Casulaties> casulatiesList = casualtiesService.list(casulatieslqw);
                if(ObjectUtils.isNotEmpty(casulatiesList) && casulatiesList.size()>0){
                    //查询不同烈度下各个建筑物类型的易损性数据
                    LambdaQueryWrapper<InfluenceYsx> lambdaQueryWrapper = new LambdaQueryWrapper<>();
                    lambdaQueryWrapper.eq(InfluenceYsx::getEarthQuakeId,earthQuakeId).eq(InfluenceYsx::getInfluenceVersion,influenceVersion);
                    List<InfluenceYsx> influenceYsxList = influenceYsxService.list(lambdaQueryWrapper);

                    if(ObjectUtils.isNotEmpty(influenceYsxList) && influenceYsxList.size()>0){
                        Map<String, List<InfluenceYsx>> intensityMap = influenceYsxList.stream().collect(Collectors.groupingBy(InfluenceYsx::getIntensity));
                        //按照模型分组
                        Map<Integer, List<Casulaties>> modelMap = casulatiesList.stream().collect(Collectors.groupingBy(Casulaties::getModelType));
                        Iterator<Map.Entry<Integer, List<Casulaties>>> iterator = modelMap.entrySet().iterator();
                        while(iterator.hasNext()){
                            Map.Entry<Integer, List<Casulaties>> next = iterator.next();
                            //模型类型  每个类型都要计算一个失去住所人数
                            Integer modelType = next.getKey();
                            //根据各个模型算法计算得到的人员伤亡数据 模型1-3一个烈度是一个伤亡值，模型4-5一个烈度可能存在多个伤亡值
                            List<Casulaties> value = next.getValue();
                            for (int i = 0; i < value.size(); i++) {
                                LossHouse lossHouse = new LossHouse();
                                Casulaties casulaties = value.get(i);
                                BigDecimal deathNum = BigDecimal.ZERO;
                                if(i == 0){
                                    deathNum = casulaties.getDeathNum();
                                }
                                Integer intensity = casulaties.getIntensity();
                                List<InfluenceYsx> influenceYsxes = intensityMap.get(intensity + "");
                                if(ObjectUtils.isEmpty(influenceYsxes) || influenceYsxes == null){
                                    continue;
                                }
                                //对于模型4和5来说因为影响场和易损性交互，可能同一个烈度圈被分成了若干部分，所以存在相同烈度对应不同的易损性数据，也就存在相同烈度有不同的相交的空间数据
                                Map<String, List<InfluenceYsx>> collect = influenceYsxes.stream().collect(Collectors.groupingBy(InfluenceYsx::getGeom));
                                BigDecimal totalLossHouseNum = BigDecimal.ZERO;
                                String geom = "";
                                //人员伤亡模型4和5计算出来的伤亡数据的空间数据、个数和失去住所的数据的空间数据、个数能一一对应的上，但是模型1-3因为没有使用易损性所以个数和空间
                                //数据对应不上就随意选择一个相同烈度的即可
                                List<InfluenceYsx> influenceYsxes1 = collect.get(casulaties.getGeom());
                                if(ObjectUtils.isEmpty(influenceYsxes1)){
                                    Iterator<Map.Entry<String, List<InfluenceYsx>>> iterator1 = collect.entrySet().iterator();
                                    while(iterator1.hasNext()){
                                        Map.Entry<String, List<InfluenceYsx>> next1 = iterator1.next();
                                        if(ObjectUtils.isEmpty(influenceYsxes1)){
                                            influenceYsxes1 = next1.getValue();
                                            break;
                                        }
                                    }
                                }
                                for (InfluenceYsx entity : influenceYsxes1) {
                                    //计算（c+d+e/2） c,d,e分别是各结构类型的建筑物易损性数据毁坏、严重破坏、中等破坏的比率
                                    BigDecimal lossHouseNum = entity.getDestory().add(entity.getSerious()).add(entity.getMid().divide(BigDecimal.valueOf(2), 4, RoundingMode.HALF_DOWN));
                                    lossHouseNum = lossHouseNum.multiply(BigDecimal.valueOf(entity.getPopNum())).subtract(deathNum);
                                    totalLossHouseNum = totalLossHouseNum.add(lossHouseNum);
                                    geom = entity.getGeom();
                                }
                                lossHouse.setId(UUID.randomUUID().toString().replace("-","")).setEarthQuakeId(earthQuakeId).setCreateTime(new Date())
                                        .setInfluenceVersion(influenceVersion).setIntensity(intensity+"").setGeom(geom).setLossHouseNum(totalLossHouseNum).setModelType(modelType);
                                resultList.add(lossHouse);
                            }
                        }
                    }
                }
                log.error("地震id"+earthQuakeId+"-- 评估失去住所人数"+"|| 成功。>> 参数:"+ JSON.toJSONString(map));
                if(resultList.size()>0){
                    saveBatch(resultList);
                } 
            }catch (Exception e){
                String msgId = earthQuakeId;
                int retryCount = retryMap.getOrDefault(msgId, 0);
                log.error("{},--评估失去住所人数||失败,处理消息时发生错误:{},即将开始第{}次消息重试...",msgId,e.getMessage(),retryCount + 1, e);
                if (retryCount >= failedRetry - 1) {
                    log.error("{},---评估失去住所人数||失败,消息将被拒绝，>>参数:{}",msgId,JSON.toJSONString(map));
                    // 拒绝消息，并且不重新入队
                    try {
                        channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
                    } catch (IOException ex) {
                        ex.printStackTrace();
                    }
                    retryMap.remove(msgId);
                } else {
                    // 重新发送消息到队列
                    try {
                        channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
                                message.getMessageProperties().getReceivedRoutingKey(),
                                MessageProperties.MINIMAL_PERSISTENT_BASIC,
                                JSON.toJSONBytes(map));
                    } catch (IOException ex) {
                        ex.printStackTrace();
                    }
                    retryMap.put(msgId, retryCount + 1);
                    log.error("地震id"+earthQuakeId+"-- 评估失去住所人数"+"|| 失败。>> 参数:"+ JSON.toJSONString(map));
                }
                e.printStackTrace();
            }finally {
                DynamicDataSourceContextHolder.clear();
            }
            log.info("失去住所人数评估结束");
        },executor);

    }

    @DSTransactional
    public void evaluation(String earthQuakeId,String influenceVersion) throws Exception{
        //先删除之前生成的相同影响场版本的数据
        LambdaQueryWrapper<LossHouse> deleteWrapper = new LambdaQueryWrapper<>();
        deleteWrapper.eq(LossHouse::getInfluenceVersion,influenceVersion).eq(LossHouse::getEarthQuakeId,earthQuakeId);
        remove(deleteWrapper);
        List<LossHouse> resultList = new ArrayList<>();
        //获取人员伤亡的结果
        LambdaQueryWrapper<Casulaties> casulatieslqw = new LambdaQueryWrapper<>();
        casulatieslqw.eq(Casulaties::getEarthQuakeId,earthQuakeId).eq(Casulaties::getVersion,influenceVersion);
        List<Casulaties> casulatiesList = casualtiesService.list(casulatieslqw);
        if(ObjectUtils.isNotEmpty(casulatiesList) && casulatiesList.size()>0){
            //查询不同烈度下各个建筑物类型的易损性数据
            LambdaQueryWrapper<InfluenceYsx> lambdaQueryWrapper = new LambdaQueryWrapper<>();
            lambdaQueryWrapper.eq(InfluenceYsx::getEarthQuakeId,earthQuakeId).eq(InfluenceYsx::getInfluenceVersion,influenceVersion);
            List<InfluenceYsx> influenceYsxList = influenceYsxService.list(lambdaQueryWrapper);

            if(ObjectUtils.isNotEmpty(influenceYsxList) && influenceYsxList.size()>0){
                Map<String, List<InfluenceYsx>> intensityMap = influenceYsxList.stream().collect(Collectors.groupingBy(InfluenceYsx::getIntensity));
                //按照模型分组
                Map<Integer, List<Casulaties>> modelMap = casulatiesList.stream().collect(Collectors.groupingBy(Casulaties::getModelType));
                Iterator<Map.Entry<Integer, List<Casulaties>>> iterator = modelMap.entrySet().iterator();
                while(iterator.hasNext()){
                    Map.Entry<Integer, List<Casulaties>> next = iterator.next();
                    //模型类型  每个类型都要计算一个失去住所人数
                    Integer modelType = next.getKey();
                    //根据各个模型算法计算得到的人员伤亡数据 模型1-3一个烈度是一个伤亡值，模型4-5一个烈度可能存在多个伤亡值
                    List<Casulaties> value = next.getValue();
                    for (int i = 0; i < value.size(); i++) {
                        LossHouse lossHouse = new LossHouse();
                        Casulaties casulaties = value.get(i);
                        BigDecimal deathNum = BigDecimal.ZERO;
                        if(i == 0){
                            deathNum = casulaties.getDeathNum();
                        }
                        Integer intensity = casulaties.getIntensity();
                        List<InfluenceYsx> influenceYsxes = intensityMap.get(intensity + "");
                        //对于模型4和5来说因为影响场和易损性交互，可能同一个烈度圈被分成了若干部分，所以存在相同烈度对应不同的易损性数据，也就存在相同烈度有不同的相交的空间数据
                        Map<String, List<InfluenceYsx>> collect = influenceYsxes.stream().collect(Collectors.groupingBy(InfluenceYsx::getGeom));
                        BigDecimal totalLossHouseNum = BigDecimal.ZERO;
                        String geom = "";
                        //人员伤亡模型4和5计算出来的伤亡数据的空间数据、个数和失去住所的数据的空间数据、个数能一一对应的上，但是模型1-3因为没有使用易损性所以个数和空间
                        //数据对应不上就随意选择一个相同烈度的即可
                        List<InfluenceYsx> influenceYsxes1 = collect.get(casulaties.getGeom());
                        if(ObjectUtils.isEmpty(influenceYsxes1)){
                            Iterator<Map.Entry<String, List<InfluenceYsx>>> iterator1 = collect.entrySet().iterator();
                            while(iterator1.hasNext()){
                                Map.Entry<String, List<InfluenceYsx>> next1 = iterator1.next();
                                if(ObjectUtils.isEmpty(influenceYsxes1)){
                                    influenceYsxes1 = next1.getValue();
                                    break;
                                }
                            }
                        }
                        for (InfluenceYsx entity : influenceYsxes1) {
                            //计算（c+d+e/2） c,d,e分别是各结构类型的建筑物易损性数据毁坏、严重破坏、中等破坏的比率
                            BigDecimal lossHouseNum = entity.getDestory().add(entity.getSerious()).add(entity.getMid().divide(BigDecimal.valueOf(2), 4, RoundingMode.HALF_DOWN));
                            lossHouseNum = lossHouseNum.multiply(BigDecimal.valueOf(entity.getPopNum())).subtract(deathNum);
                            totalLossHouseNum = totalLossHouseNum.add(lossHouseNum);
                            geom = entity.getGeom();
                        }
                        lossHouse.setId(UUID.randomUUID().toString().replace("-","")).setEarthQuakeId(earthQuakeId).setCreateTime(new Date())
                                .setInfluenceVersion(influenceVersion).setIntensity(intensity+"").setGeom(geom).setLossHouseNum(totalLossHouseNum).setModelType(modelType);
                        resultList.add(lossHouse);
                    }
                }
            }
        }
        if(resultList.size()>0){
            saveBatch(resultList);
        }
    }
}
